package net.bwie.realtime.vehicle2.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.vehicle2.bean.VehicleData;
import net.bwie.realtime.vehicle2.function.VehicleAbnormalSlarmFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 异常检测与告警
 *
 * @Author: FuHe
 * @Date: 2025/6/7
 */
public class VehicleAbnormalAlarmJob2 {
    public static void main(String[] args) throws Exception {
        // 1、设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //  2、读取数据 kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle-data-log");
//        kafkaStream.print();
        // 3、处理数据
        DataStream<String> processStream = handle(kafkaStream);
        processStream.print();
        // 4、写入数据 clickhouse
        JdbcUtil.sinkToClickhouseUpsert(processStream,
                "INSERT INTO vehicle_data.vehicle_abnormal_alarm (window_start, window_end, vin, alarm_type)\n" +
                        "VALUES (?, ?, ?, ?)");
        // 5、执行程序
        env.execute("VehicleAbnormalAlarmJob2");
    }

    private static DataStream<String> handle(DataStream<String> kafkaStream) {
        // 将数据流中的数据转换为VehicleData对象
        SingleOutputStreamOperator<VehicleData> map = kafkaStream.map(o -> JSON.parseObject(o, VehicleData.class));
        // 为数据流添加水印
        SingleOutputStreamOperator<VehicleData> watermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<VehicleData>() {
                    @Override
                    public long extractTimestamp(VehicleData element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                })
        );
        // 按照车辆标识符进行分组
        KeyedStream<VehicleData, String> vinStream = watermarks.keyBy(VehicleData::getVin);
        // 定义电池最高温度阈值
        final Double MaxBatteryTemp = 60.0;
        // 定义匹配模式，用于检测连续三次电池温度超过阈值的事件
        Pattern<VehicleData, VehicleData> pattern = Pattern.<VehicleData>begin("a")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) throws Exception {
                        // 检查电池温度是否非空且超过最大值
                        return value.getBatteryTemp() != null
                                && value.getBatteryTemp() > MaxBatteryTemp;
                    }
                })
                .next("b")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) throws Exception {
                        // 再次检查电池温度是否非空且超过最大值
                        return value.getBatteryTemp() != null
                                && value.getBatteryTemp() > MaxBatteryTemp;
                    }
                }).next("c")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) throws Exception {
                        return value.getBatteryTemp() != null
                                && value.getBatteryTemp() > MaxBatteryTemp;
                    }
                })
                .within(Time.minutes(15));

        // 创建模式流
        PatternStream<VehicleData> patternStream = CEP.pattern(vinStream, pattern);
        // 处理匹配到的模式并生成报警信息
        SingleOutputStreamOperator<String> restStream = patternStream.process(
                new PatternProcessFunction<VehicleData, String>() {
                    @Override
                    public void processMatch(Map<String, List<VehicleData>> match, Context ctx, Collector<String> out) throws Exception {
                        // 获取匹配结果中的第一个车辆数据
                        VehicleData a = match.get("a").get(0);
                        VehicleData b = match.get("b").get(0);
                        VehicleData c = match.get("c").get(0);

                        // 将时间戳转换为可读的时间格式
                        String startTime = DateTimeUtil.convertLongToString(a.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);
                        String endTime = DateTimeUtil.convertLongToString(c.getTimestamp(), DateTimeUtil.DATE_TIME_FORMAT);

                        // 构造报警信息
                        String alarmStream = String.join(",",
                                startTime,
                                endTime,
                                a.getVin(),
                                "连续三次高温报警");

                        // 输出报警信息
                        out.collect(alarmStream);

                    }
                }
        );
        SingleOutputStreamOperator<String> process = vinStream.process(new VehicleAbnormalSlarmFunction());
        DataStream<String> union = process.union(restStream);
        return union;
    }
}
